-- **************************************************************************
--    Project Name:   干线车辆操作时效监控报表-明细数据
--    Job Name:       jms_dm__dm_tms_trunk_oper_eff_dt
--    Description :   计算每条干线在每个站点对每个运单的操作时间
--    Author :        禹铄
--    date：          2021/11/17
-- **************************************************************************
--    Relation ：
--    jms_dm.dm_tms_trunk_oper_eff_dt << [
--        jms_ods.yl_tmsnew_tms_shipment_stop,
--        jms_ods.yl_tmsnew_tms_shipment,
--        jms_ods.tab_barscan_loading,
--        jms_ods.tab_barscan_unloading
--    ]
-- **************************************************************************
--    modify by 禹铄 2021/3/16 迁移表从JMS_DWD 到 JMS_DM 层
-- **************************************************************************

set hive.exec.dynamic.partition=true;            --  动态分区
set hive.exec.dynamic.partition.mode=nonstrict;
set hive.exec.max.dynamic.partitions=500;         --  每天生成 60 个分区
set hive.exec.max.dynamic.partitions.pernode=500; --  每天生成 60 个分区
with tmp as (select c.shipment_no,                                      -- 取出所有干线任务 将干线任务拆分 A-B-C  拆解成 A-B B-C
                    c.shipment_name,
                    concat(c.setout_sitename, '-', c.arrive_sitename) setout_arrive_sitename,
                    c.setout_sitename,
                    c.setout_sitecode,
                    c.setout_actual_arivtime,
                    c.arrive_sitename,
                    c.arrive_sitecode,
                    c.arrive_actual_arivtime
             from (
                      select a.shipment_no,-- 任务编号
                             b.shipment_name,
                             a.network_code                                             as setout_sitecode,
                             a.network_name                                             as setout_sitename,
                             a.actual_arrival_time                                      as setout_actual_arivtime,
                             lead(a.network_code, 1, null)
                                  over ( partition by a.shipment_no order by sort_num ) as arrive_sitecode,
                             lead(a.network_name, 1, null)
                                  over (partition by a.shipment_no order by sort_num)   as arrive_sitename,
                             lead(a.actual_arrival_time, 1, null)
                                  over (partition by a.shipment_no order by sort_num)   as arrive_actual_arivtime
                      from (select * from jms_dwd.dwd_tmsnew_shipment_stop_union_base_dt where dt >=date_add('{{ execution_date | cst_ds }}',-7) and table_type=1) a -- 干线经停表
                               join (select * from jms_dwd.dwd_tmsnew_shipment_union_base_dt where dt >=date_add('{{ execution_date | cst_ds }}',-7) and table_type=1)  b -- 干线任务表
                                    on a.shipment_no = b.shipment_no
                  ) c
             where c.arrive_sitename is not null
                  and c.arrive_actual_arivtime is not null
),
     loading_rk as (                                                            -- 装车表取T-3天进行去重（取最大装车时间）
         select tmp_rk.billcode,
                tmp_rk.scansitecode,
                tmp_rk.transfercode,
                tmp_rk.nextstation,
                tmp_rk.scantime as load_scantime
         from (
                  select n.waybill_no billcode, -- 运单编号
                         n.scan_site_code scansitecode, -- 扫描网点编号
                         n.shipment_no transfercode, -- 转单号
                         n.next_station nextstation, -- 上一站或下一站
                         n.scan_time scantime, -- 扫描时间
                         row_number()
                                 over (partition by n.waybill_no, n.scan_site_code , n.shipment_no  order by n.scan_time desc ) rk
                  from jms_dwd.dwd_tab_barscan_loading_base_dt n
                  where n.dt <= '{{ execution_date | cst_ds }}'
                    and n.dt > date_add('{{ execution_date | cst_ds }}', -3)
             ) tmp_rk
         where rk = 1
     ),
     -- 卸车
     unloading_rk as (                                                              -- 卸车表取T-3天进行去重（取出最小卸车时间）
         select tmp_rk.billcode,
                tmp_rk.scansitecode,
                tmp_rk.packagecode,
                tmp_rk.transfercode,
                tmp_rk.scantime as unload_scantime
         from (
                  select a.waybill_no billcode,  -- 运单编号
                         a.scan_site_code scansitecode, -- 扫描网点编号
                         a.package_code packagecode,  -- 所属袋号
                         a.scan_time scantime,  -- 扫描时间
                         a.shipment_no transfercode, -- 转单号
                         row_number()
                                 over (partition by a.waybill_no,a.scan_site_code,a.shipment_no order by a.scan_time ) rk
                  from jms_dwd.dwd_tab_barscan_unloading_base_dt a
                  where a.dt <= '{{ execution_date | cst_ds }}'
                    and a.dt > date_add('{{ execution_date | cst_ds }}', -3)
             ) tmp_rk
         where rk = 1
     )

insert overwrite table jms_dm.dm_tms_trunk_oper_eff_dt partition(dt)
select t1.shipment_no,                                                                                               -- 任务号
       t1.shipment_name,                                                                                             -- 任务名字
       t1.setout_arrive_sitename,                                                                                    -- 发出及到达站点名
       t1.setout_sitename,                                                                                           -- 发出站点名字
       t1.setout_sitecode,                                                                                           -- 发出站点代码
       t1.arrive_sitename,                                                                                           -- 到站站点名字
       t1.arrive_sitecode,                                                                                           -- 到站站点代码
       nvl(t1.packagecode, '--')                                                              as packagecode,        -- 包号
       t2.nextstation                                                                         as nextstation,        -- 下一站点名称
       t1.billcode                                                                            as billcode,           -- 单号
       t1.unload_scantime                                                                     as arrive_scantime,    -- 卸车扫描时间
       t2.load_scantime                                                                       as setout_scantime,    -- 装车扫描时间
       round((unix_timestamp(t2.load_scantime) - unix_timestamp(t1.unload_scantime)) / 60, 2) as operation_duration, -- 操作时间
       cast(t1.arrive_actual_arivtime as date)                                                as actual_arrival_time,-- 实际到车时间
       cast(t1.arrive_actual_arivtime as date)                                                as dt                  -- 分区字段
from (
         select tmp.shipment_no,
                tmp.shipment_name,   -- 任务    名字及号码
                tmp.setout_arrive_sitename,
                tmp.setout_sitename,
                tmp.setout_sitecode, -- 发件中心 名字及代码
                tmp.arrive_sitename,
                tmp.arrive_sitecode, -- 到件中心 名字及代码
                m.billcode,
                m.packagecode,       -- 单和包号
                m.unload_scantime,   -- 卸车扫描时间
                tmp.arrive_actual_arivtime
         from unloading_rk m
                  join tmp
                       on tmp.shipment_no = m.transfercode          -- 任务号 和 转单号
                           and tmp.arrive_sitecode = m.scansitecode -- 到达站点编号 和 扫描站点编号
     ) t1
         left join loading_rk t2
                   on t1.billcode = t2.billcode                 -- 运单号
                       and t1.arrive_sitecode = t2.scansitecode -- 到达站点
where date_format(t1.arrive_actual_arivtime,'yyyy-MM-dd') = date_add('{{ execution_date | cst_ds }}',-1)
group by t1.shipment_no,
         t1.shipment_name,
         t1.setout_arrive_sitename,
         t1.setout_sitename,
         t1.setout_sitecode,
         t1.arrive_sitename,
         t1.arrive_sitecode,
         t1.packagecode,
         t2.nextstation,
         t1.billcode,
         t1.unload_scantime,
         t2.load_scantime,
         t1.arrive_actual_arivtime
distribute by pmod(hash(rand()),21) ;